package datastore.api.impl.hbase;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.io.BatchUpdate;

import datastore.db.Key;

public abstract class BaseStore<R, P> implements Store<R, P> {

    protected static final byte[] EMPTY_VALUE = new byte[] {};

    protected HBaseDatastore hbase;
    protected HBaseConfiguration config;
    protected HBaseAdmin admin;
    protected HTable table;
    protected String appName;

    public BaseStore(HBaseDatastore hbase) {
        this.hbase = hbase;
        this.config = hbase.config;
        this.admin = hbase.admin;
        this.appName = hbase.appName;
    }

    protected void createTable(List<BatchUpdate> updates, String tableName, String columnFamily,
            String column, String defaultRowKey) throws IOException {
        HTableDescriptor desc = new HTableDescriptor(tableName);
        desc.addFamily(new HColumnDescriptor(columnFamily));
        admin.createTable(desc);
        updates.add(getBatch(defaultRowKey, column, EMPTY_VALUE));
    }

    protected BatchUpdate getBatch(String rowKey, String column, byte[] value) {
        BatchUpdate update = new BatchUpdate(rowKey);
        update.put(column, value);
        return update;
    }

    protected void putRow(String rowKey, String tableName, String column, byte[] value)
            throws IOException {
        BatchUpdate update = getBatch(rowKey, column, value);
        table.commit(update);
    }

    protected void deleteRow(String rowKey, String tableName) throws IOException {
        table.deleteAll(rowKey);
    }

    public String toEntityRowKey(Key key) {
        if (key == null) {
            throw new IllegalArgumentException("key cannot be null");
        }
        return constructKey(key.getKind().getName(), key.getIdOrName());
    }
    
    protected String constructKey(Object... parts) {
        StringBuilder key = new StringBuilder(appName);
        for (Object part : parts) {
            if (part == DEFAULT_KEY_PART) {
                key.append(" "); // for start rows
            } else {
                key.append(KEY_DELIM);
            }
            key.append(part);
        }
        return key.toString();
    }

    public Key toKey(String entityRowKey) throws Exception {
        String[] split = entityRowKey.split(KEY_DELIM);
        String[] values = Arrays.copyOfRange(split, 1, split.length); //idx 0 is appName
        Object[] args = new Object[values.length];
        for (int i = 0; i < args.length; i++) {
            if (i % 2 == 0) { //even elements in rowKeys are Model kinds, odd are idOrName's
                args[i] = Class.forName(values[i]);
            } else if (Character.isDigit(values[i].charAt(0))) { //if it starts with a number, id
                args[i] = new Long(values[i]);
            } else { // otherwise name
                args[i] = values[i];
            }
        }
        return Key.fromPath(args);
    }

    public static String escape(String rowKey) {
        return rowKey.replace(KEY_DELIM, ESCAPED_DELIM);
    }

    public static String unescape(String rowKey) {
        return rowKey.replace(ESCAPED_DELIM, KEY_DELIM);
    }
    
    public static String getLastPart(String rowKey) {
        String[] parts = rowKey.split(Store.KEY_DELIM);
        return parts[parts.length - 1];
    }
}